package com.haiyou.common.thread.pool;

import java.util.NoSuchElementException;
import java.util.Queue;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import com.google.common.collect.Queues;
import com.haiyou.common.thread.task.AbstractOrderedTask;
import com.haiyou.common.thread.task.Busilable;
import com.haiyou.common.util.hash.HashUtils;
import com.haiyou.common.util.string.StringUtils;
import com.haiyou.common.util.system.SystemUtils;

import lombok.extern.slf4j.Slf4j;

/**
 * 
 * @Description: 线程安全线程池 ： Actor模型
 * @author xingyuan
 * @date 2023年5月5日 上午11:22:45
 */
@Slf4j
final public class FastActorThreadPoolExecutor implements Executor, Busilable {

	final String _threadNamedPrefix;
	final FastThread[] _ts;

	public FastActorThreadPoolExecutor(int nThreads, String threadNamedPrefix) {
		_threadNamedPrefix = threadNamedPrefix;
		_ts = new FastThread[nThreads];
		for (int i = 0; i < _ts.length; i++) {
			_ts[i] = new FastThread(StringUtils.merge(threadNamedPrefix, i), busily());
		}
		log.warn("{} [{}] starting!!!", getClass().getSimpleName(), threadNamedPrefix);
	}
	
	public long[] threadCount() {
		int length = _ts.length;
		long[] result = new long[length * 2];
		for (int i = 0; i < length; i++) {
			result[2 * i] = _ts[i]._queue._size;
			result[2 * i + 1] = _ts[i].count.get();
		}
		return result;
	}

	private void shutdown() {
		for (FastThread t : _ts) {
			t.shutdown();
		}
		log.warn("{} [{}] stopping!!!", getClass().getSimpleName(), _threadNamedPrefix);
	}

	private void shutdownNow() {
		for (FastThread t : _ts) {
			t.shutdownNow();
		}
	}

	private boolean isTerminated() {
		for (FastThread t : _ts) {
			if (t.isAlive())
				return false;
		}
		return true;
	}

	private void doRemaining() {
		Queue<Runnable> q = Queues.newConcurrentLinkedQueue();
		for (FastThread t : _ts) {
			while (t._queue._size > 0) {
				q.offer(t._queue.poll());
			}
		}
		while (q.size() > 0) {
			Runnable r;
			try {
				r = q.poll();
				r.run();
			} catch (Exception e) {
				log.error(e.getMessage(), e);
			}
		}
	}

	public void shutdownNowGracefully() {
		shutdownNow();
		while (!isTerminated())
			SystemUtils.sleep(1, TimeUnit.SECONDS);
		doRemaining();
		log.warn("{} [{}] stopped!!!", getClass().getSimpleName(), _threadNamedPrefix);
	}

	public void shutdownGracefully() {
		shutdown();
		while (!isTerminated())
			SystemUtils.sleep(1, TimeUnit.SECONDS);
		log.warn("{} [{}] stopped!!!", getClass().getSimpleName(), _threadNamedPrefix);
	}

	public <Actor> void addTask(Actor actor, Runnable task) {
		int index = 0;
		if (_ts.length > 1)
			index = HashUtils.hashCode(actor) % _ts.length;
		FastThread fastThread = _ts[index];
		fastThread.execute(task);
	}

	@Override
	public void execute(Runnable command) {
		if (command instanceof AbstractOrderedTask) {
			addTask(((AbstractOrderedTask<?>) command).getSer(), command);
		} else {
			addTask(command, command);
		}
	}
	
	
	private <Actor> void addTaskWithMaxLength(Actor actor, Runnable task, int maxLength) {
		int index = 0;
		if (_ts.length > 1)
			index = HashUtils.hashCode(actor) % _ts.length;
		FastThread fastThread = _ts[index];
		if (fastThread._queue._size < maxLength) {
			fastThread.execute(task);
		}
	}

	
	public void executeWithMaxLengthCheck(Runnable command, int maxLength) {
		if (command instanceof AbstractOrderedTask) {
			addTaskWithMaxLength(((AbstractOrderedTask<?>) command).getSer(), command, maxLength);
		} else {
			addTaskWithMaxLength(command, command, maxLength);
		}
	}

}

final class Node<E> {
	E _item;
	Node<E> _next;

	Node(E element, Node<E> next) {
		this._item = element;
		this._next = next;
	}
}

final class FastQueue<E> {

	int _size = 0;
	Node<E> _first;
	Node<E> _last;

	void offer(E e) {
		final Node<E> l = _last;
		final Node<E> newNode = new Node<>(e, null);
		_last = newNode;
		if (l == null)
			_first = newNode;
		else
			l._next = newNode;
		_size++;
	}

	E poll() {
		final Node<E> f = _first;
		if (f == null)
			throw new NoSuchElementException();
		final E element = f._item;
		final Node<E> next = f._next;
		f._item = null;
		f._next = null;
		_first = next;
		if (next == null)
			_last = null;
		_size--;
		return element;
	}

}

@Slf4j
final class FastThread implements Runnable, Executor {

	ReentrantLock _lock = new ReentrantLock();
	Condition _condition = _lock.newCondition();
	
	AtomicLong count = new AtomicLong();
	int _backlog;
	FastQueue<Runnable> _queue;
	Thread _thread;
	boolean _shutdown;
	boolean _shutdownNow;

	FastThread(String name, int backlog) {
		_backlog = backlog;
		_queue = new FastQueue<>();
		_thread = new Thread(this, name);
		_thread.start();
	}

	boolean isAlive() {
		return _thread.isAlive();
	}

	void shutdown() {
		final ReentrantLock lock = this._lock;
		try {
			lock.lockInterruptibly();
			_shutdown = true;
			_condition.signal();
		} catch (InterruptedException e) {
			log.error(e.getMessage(), e);
		} finally {
			lock.unlock();
		}
	}

	void shutdownNow() {
		final ReentrantLock lock = this._lock;
		try {
			lock.lockInterruptibly();
			_shutdownNow = true;
			_condition.signal();
		} catch (InterruptedException e) {
			log.error(e.getMessage(), e);
		} finally {
			lock.unlock();
		}
	}

	@Override
	public void execute(Runnable command) {
		final ReentrantLock lock = this._lock;
		try {
			lock.lockInterruptibly();
			_queue.offer(command);
			_condition.signal();
		} catch (InterruptedException e) {
			log.error(e.getMessage(), e);
		} finally {
			lock.unlock();
		}
	}

	@Override
	public void run() {

		FastQueue<Runnable> exeQueue = null;
		
		for (;;) {
			final ReentrantLock lock = this._lock;
			try {
				lock.lockInterruptibly();
				if (_shutdownNow) {
					break;
				}
				if (_shutdown) {
					if (_queue._size == 0) {
						break;
					}
					exeQueue = _queue;
					_queue = new FastQueue<>();
				} else {
					if (_queue._size == 0) {
						try {
							_condition.await();
						} catch (InterruptedException e) {
							log.error(e.getMessage(), e);
							break;
						}
					}
					exeQueue = _queue;
					_queue = new FastQueue<>();
				}
			} catch (InterruptedException e) {
				log.error(e.getMessage(), e);
				break;
			} finally {
				lock.unlock();
			}
			
			if (exeQueue != null) {
				count.addAndGet(exeQueue._size);
				if (exeQueue._size >= _backlog) {
					log.warn("BUSILY:taskCount=[{}],thread=[{}]", exeQueue._size, Thread.currentThread().getName());
				}
				while (exeQueue._size > 0) {
					Runnable r = null;
					try {
						r = exeQueue.poll();
						r.run();
					} catch (Exception e) {
						log.error(e.getMessage() + ":" + r, e);
					}
				}
			}

		}

	}

}
